package com.flrjcx;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * kafka消费者
 *
 * @author Flrjcx
 */
public class ConsumerKafka {
    public static void main(String[] args) {
//      1. 创建kafka消费者配置信息
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.14.157:9092");
//      消费者组（可以使用消费者组将若干个消费者组织到一起），共同消费Kafka中topic的数据
//      每一个消费者需要指定一个消费者组，如果消费者的组名是一样的，表示这几个消费者是一个组中的
        props.setProperty("group.id","test");
//      自动提交offset
        props.setProperty("enable.auto.commit","true");
//      自动提交offset时间间隔
        props.setProperty("auto.commit.interval.ms","1000");
//        拉取key,value数据
        props.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

//        2.创建消费者
        KafkaConsumer<String ,String> kafkaConsumer = new KafkaConsumer<>(props);

//        3. 订阅消费的topic
//        指定消费者从哪个topic拉取数据
        kafkaConsumer.subscribe(Arrays.asList("test"));

//        4. 死循环获取数据
        while (true){
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(5));
//        5. 将记录的offset key value都打印
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
//                主题
                String topic = consumerRecord.topic();
//                offset: 这条消息处于kafka分区中的哪个位置
                long offset = consumerRecord.offset();
//                key/value
                String key = consumerRecord.key();
                String value = consumerRecord.value();

                System.out.println("topic:"+topic+",offset:"+offset+",key:"+key+",value:"+value);
            }
        }
    }

}
